package cn.com.lyb.flink.aggreagte;

import cn.com.lyb.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * reduce可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。
 */
public class ReduceDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 3),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s3", 3L, 7),
                new WaterSensor("s3", 3L, 3));
        // 按照id 分组
        // keyby 不是转换算子，智能对数据进行重分区，不能设置并行度
        // keyby 分组与分区的关系
        // 1） keyby 是对数据分组， 保证相同key的数据再同一个分区
        // 2）分区：一个子任务，可以理解为一个分区，一个分区中可以存在多个分组（key）
        KeyedStream<WaterSensor, String> keyByDS = sensorDS.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.getId();
            }
        });
        // 只能再keyBy后调用
        // 输入类型 = 输出类型， 类型不能变
        // 每个key 的第一条数据来的时候，不会执行reduce，会存起来，直接输出
        // reduce方法的两个参数
        // v1 之前的计算结果，存状态
        // v2 现在来的数据
        keyByDS.reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor v1, WaterSensor v2) throws Exception {
                System.out.println("=======");
                return new WaterSensor(v1.getId(), v2.getTs(), v1.getVc() + v2.getVc());
            }
        }).print();

        env.execute();
    }
}
